
package org.sopt.dev.ImgKnight.agent;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.URL;
import java.text.DecimalFormat;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

import org.sopt.dev.ImgKnight.agent.Config.AgentConfig;
import org.sopt.dev.ImgKnight.agent.Controller.TelnetController;
import org.sopt.dev.ImgKnight.agent.Executor.TakeResource;
import org.sopt.dev.ImgKnight.agent.Executor.TakeResourceCompletionHandler;
import org.sopt.dev.ImgKnight.agent.Executor.TakeUrl;
import org.sopt.dev.ImgKnight.agent.Executor.TakeUrlCompletionHandler;
import org.sopt.dev.ImgKnight.agent.Filter.ImgDupFilter;
import org.sopt.dev.ImgKnight.agent.HttpHandler.HttpHandler;
import org.sopt.dev.ImgKnight.agent.NIO.NioObjectStreamHandler;
import org.sopt.dev.ImgKnight.agent.RobotsExclusion.DefaultCacheEngine;
import org.sopt.dev.ImgKnight.agent.RobotsExclusion.LRUCacheEngine;
import org.sopt.dev.ImgKnight.agent.RobotsExclusion.RobotsExclusion;
import org.sopt.dev.ImgKnight.agent.RobotsExclusion.RobotsTxtCacheEngine;
import org.sopt.dev.ImgKnight.agent.database.HarvestVO;

public class Agent {
	
	public static final int REQJOIN = 0xFDFDFDFD;
	
	public static final int REQDROP = 0xFEFEFEFE;
	
	public static final int SHUTDOWN = 0xFFFFFFFF;
	
	private AgentConfig agentConfig = null;
	
	private TelnetController telnetController = null;
	
	private HttpHandler httpHandler = null;

	private RobotsExclusion robotsExclusion = null;
	
	private Channel channel = null;
	
	private ExecutorService nioBossExecutor = null;
	
	private ExecutorService nioWorkerExecutor = null;
	
	private ExecutorService takeUrlExecutor = null;
	
	private ExecutorService takeResourceExecutor = null;

	private CompletionService<List<String>> takeUrlcompletionService = null;
	
	private CompletionService<List<HarvestVO>> takeResourceCompletionService = null;

	private Thread takeUrlCompletionHandler = null;
	
	private Thread takeResourceCompletionHandler = null;
	
	private ImgDupFilter imgDupFilter = null;
	
	private AtomicLong reservedUrlCount = null;
	
	private AtomicLong reservedResourceCount = null;
	
	private AtomicLong takeUrlCount = null;
	
	private AtomicLong discardUrlCount = null;
	
	private AtomicLong missUrlCount = null;
	
	private AtomicLong takeResourceCount = null;
	
	private AtomicLong discardResouceCount = null;
	
	private AtomicLong missResouceCount = null;

	private Logger serviceLogger = null;

	private Logger exceptionLogger = null;
	
	private Agent() {
		serviceLogger = Logger.getLogger("agent.service");
		exceptionLogger = Logger.getLogger("agent.exception");
	}

	public void init() throws IOException, FileNotFoundException {
		agentConfig  = new AgentConfig();
		agentConfig.BuildConfig(new File("agent.properties"));
		
		telnetController = new TelnetController();
		
		RobotsTxtCacheEngine<String, String[]> cacheEngine = null;
		if( agentConfig.AgentCacheEngineType.equalsIgnoreCase("LRU") )
			cacheEngine = new LRUCacheEngine(agentConfig.AgentCacheSize);
		else
			cacheEngine = new DefaultCacheEngine(agentConfig.AgentCacheSize);
		robotsExclusion = new RobotsExclusion(agentConfig.AgentName, cacheEngine);
		
		takeUrlExecutor = Executors.newFixedThreadPool(agentConfig.TakeUrlParallelDegree);
		takeUrlcompletionService = new ExecutorCompletionService<List<String>>(takeUrlExecutor);
		takeUrlCompletionHandler = Executors.defaultThreadFactory().newThread(new TakeUrlCompletionHandler());
		
		takeResourceExecutor = Executors.newFixedThreadPool(agentConfig.TakeResourceParallelDegree);
		takeResourceCompletionService = new ExecutorCompletionService<List<HarvestVO>>(takeResourceExecutor);
		takeResourceCompletionHandler = Executors.defaultThreadFactory().newThread(new TakeResourceCompletionHandler());
		
		imgDupFilter = new ImgDupFilter(agentConfig.SeenImageFilterHighWaterMark, false);
		
		reservedUrlCount = new AtomicLong(0);
		reservedResourceCount = new AtomicLong(0);
		takeUrlCount = new AtomicLong(0);
		discardUrlCount = new AtomicLong(0);
		missUrlCount = new AtomicLong(0);
		takeResourceCount = new AtomicLong(0);
		discardResouceCount = new AtomicLong(0);
		missResouceCount = new AtomicLong(0);
		httpHandler = new HttpHandler();
		
		File downloadBaseDir = new File(agentConfig.DownloadBaseDir);
		downloadBaseDir.mkdirs();
		serviceLogger.info("agent init");
	}
	
	public void run() throws IllegalArgumentException {
		nioBossExecutor = Executors.newCachedThreadPool();
		nioWorkerExecutor = Executors.newCachedThreadPool();
		ChannelFactory factory 
			= new NioClientSocketChannelFactory(nioBossExecutor, nioWorkerExecutor);	
		ClientBootstrap bootstrap = new ClientBootstrap(factory);
		NioObjectStreamHandler handler = new NioObjectStreamHandler();
		
		bootstrap.getPipeline().addLast("handler", handler);
		bootstrap.setOption("tcpNoDelay", true);
		bootstrap.setOption("keepAlive", true);
		
		SocketAddress socketAddress = new InetSocketAddress(agent.getAgentConfig().URLFrontierServIp, 
				agent.getAgentConfig().URLFrontierServPort);
		ChannelFuture future = bootstrap.connect(socketAddress);
		channel = future.awaitUninterruptibly().getChannel();
		
		if( !future.isSuccess() )
			throw new IllegalArgumentException(future.getCause().getMessage() +" server: "+
					agent.getAgentConfig().URLFrontierServIp+":" +
					agent.getAgentConfig().URLFrontierServPort);
		
		takeUrlCompletionHandler.start();
		takeResourceCompletionHandler.start();
		telnetController.open();
		serviceLogger.info("agent run");
	}
	
	public void shutdown(){
		nioBossExecutor.shutdownNow();
		nioWorkerExecutor.shutdownNow();
		takeUrlExecutor.shutdownNow().clear();
		takeResourceExecutor.shutdownNow().clear();
		channel.close().awaitUninterruptibly();
		telnetController.close();
		takeUrlCompletionHandler.interrupt();
		takeResourceCompletionHandler.interrupt();
		serviceLogger.info("agent shutdown");
	}
	
	public void onConnectionEstablished(Channel channel) {
		serviceLogger.info("Connect accepted: "+channel.getRemoteAddress().toString());
		channel.write(new Integer(Agent.REQJOIN));
	}
	
	public void onConnectionClosed(Channel channel) {
		serviceLogger.error("Connection closed: "+channel.getRemoteAddress().toString());
		shutdown();
	}
	
	@SuppressWarnings("unchecked")
	public void onRecvedObject(Object obj) {
		if( obj instanceof Map ) {
			Map pair = (Map)obj;
			String countryCode = (String)pair.get("countryCode");
			String countryName = (String)pair.get("countryName");
			URL url = (URL)pair.get("URL");
			if( robotsExclusion(url) ) {
				TakeUrl takeUrl = new TakeUrl(url, countryCode, countryName);
				takeUrlcompletionService.submit(takeUrl);
				reservedUrlCount.incrementAndGet();
			}
		}
		else if( obj instanceof List ) {
			List<Map> urls = (List<Map>)obj;
			for(Map pair : urls) {
				String countryCode = (String)pair.get("countryCode");
				String countryName = (String)pair.get("countryName");
				URL url = (URL)pair.get("URL");
				if( robotsExclusion(url) ) {
					TakeUrl takeUrl = new TakeUrl(url, countryCode, countryName);
					takeUrlcompletionService.submit(takeUrl);
					reservedUrlCount.incrementAndGet();
				}
			}
		}
		else if( obj instanceof Integer ) {
			Integer r = (Integer)obj;
			int signum = r.intValue();
			if( signum == SHUTDOWN ) {
				serviceLogger.warn("*** Shutdown signal ***");
				shutdown();
			}
		}
	}
	
	public void onCrawled(List<String> extracted) {
		takeUrlCount.incrementAndGet();
		if( extracted.size() > 0 )
			channel.write(extracted);
	}
	
	public void onDiscardCrawl() {
		discardUrlCount.incrementAndGet();
	}
	
	public void onMissedCrawl() {
		missUrlCount.incrementAndGet();
	}
	
	public void onDownloaded(List<HarvestVO> downloaded) {
		takeResourceCount.addAndGet(downloaded.size());
	}
	
	public void onDiscardDownload(int discardCount) {
		discardResouceCount.addAndGet(discardCount);
	}
	
	public void onMissedDownload(int missCount) {
		missResouceCount.addAndGet(missCount);
	}
	
	static Runtime runtime = Runtime.getRuntime();
	static DecimalFormat df = new DecimalFormat(",##0.00");
	public String getCondition() {
		return new String(
				(df.format((float)runtime.maxMemory()/1000000))+":"+
				(df.format((float)runtime.totalMemory()/1000000))+":"+
				(df.format((float)runtime.freeMemory()/1000000))+":"+
				reservedUrlCount.get()+":"+
				takeUrlCount.get()+":"+
				discardUrlCount.get()+":"+
				missUrlCount.get()+":"+
				reservedResourceCount.get()+":"+
				takeResourceCount.get()+":"+
				discardResouceCount.get()+":"+
				missResouceCount.get()+":"+
				imgDupFilter.getDuplicatedImgCount()+":"+
				imgDupFilter.getSeenImageCurrentSize()+":"+
				imgDupFilter.getSeenImageFlushedSize());
	}
	
	public void startTakeResource(List<String> extracted, String countryCode, String countryName) {
		reservedResourceCount.addAndGet(extracted.size()-1);
		TakeResource takeImage = new TakeResource(extracted, countryCode, countryName);
		try {
			takeResourceCompletionService.submit(takeImage);
		} catch (RejectedExecutionException e) {
			exceptionLogger.error("RejectedExecutionException: "+e.getMessage());
		}
	}
	
	public boolean robotsExclusion(URL url) {
		if( agentConfig.UsedAgentExclusion && robotsExclusion.disallowd(url) ) {
			serviceLogger.info("Disallowd: " + url.toString());
			return false;
		}
		return true;
	}
	
	public ImgDupFilter getImgDupFilter() {
		return imgDupFilter;
	}
	
	public long getTakeUrlCount() {
		return takeUrlCount.get();
	}
	
	public TelnetController getTelnetController() {
		return telnetController;
	}
	
	public AgentConfig getAgentConfig() {
		return agentConfig;
	}
	
	public RobotsExclusion getRobotsExclusion() {
		return robotsExclusion;
	}
		
	public HttpHandler getHttpHandler() {
		return httpHandler;
	}
	
	public CompletionService<List<String>> getTakeUrlcompletionService() {
		return takeUrlcompletionService;
	}
	
	public CompletionService<List<HarvestVO>> getTakeResourceCompletionService() {
		return takeResourceCompletionService;
	}
	
	private static Agent agent = null;
	public static Agent getInstance() {
		if( agent == null )
			agent = new Agent();
		return agent;
	}
}
